package org.jmetrix.throughput;

import org.jmetrix.api.Aggregator;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * @author Konstantin Kuzmin
 * @version $Revision$
 * @since 05.09.12
 */
public class ThroughputAggregatorImpl implements ThroughputAggregator, Aggregator<ThroughputEvent> {
    protected static final Map<TimeUnit, String> TIME_UNIT_NAMES = new HashMap<TimeUnit, String>();

    static {
        TIME_UNIT_NAMES.put(TimeUnit.NANOSECONDS, "nanosec");
        TIME_UNIT_NAMES.put(TimeUnit.MICROSECONDS, "microsec");
        TIME_UNIT_NAMES.put(TimeUnit.MILLISECONDS, "millisec");
        TIME_UNIT_NAMES.put(TimeUnit.SECONDS, "sec");
        TIME_UNIT_NAMES.put(TimeUnit.MINUTES, "min");
        TIME_UNIT_NAMES.put(TimeUnit.HOURS, "hour");
        TIME_UNIT_NAMES.put(TimeUnit.DAYS, "day");
    }

    long previousCount = 0;
    long count = 0;
    long firstEventTime = -1;
    long aggregationIntervalStartTime = -1;
    TimeUnit timeUnit;
    private long nanosInEventTimeUnit;

    @Override
    public synchronized void aggregate(ThroughputEvent measurement) {
        if (measurement.getValue() < aggregationIntervalStartTime) {
            // ignore "stale" measurements belong to the previous interval
            return;
        }
        if (firstEventTime == -1) { // it is first measurement
            firstEventTime = measurement.getValue();
            aggregationIntervalStartTime = firstEventTime;
            timeUnit = measurement.getTimeUnit();
            nanosInEventTimeUnit = timeUnit.toNanos(1);
            count = 0;
        } else {
            long timeUnitsPassed = (measurement.getValue() - aggregationIntervalStartTime) / nanosInEventTimeUnit;
            if (timeUnitsPassed > 0) {
                aggregationIntervalStartTime += timeUnitsPassed * nanosInEventTimeUnit;
                if (timeUnitsPassed == 1) {
                    previousCount = count;
                } else {
                    previousCount = 0;
                }
                count = 0;
            }
        }
        count++;
    }

    @Override
    public void doReset() {
        reset();
    }

    @Override
    public void reset() {
        firstEventTime = -1;
    }

    @Override
    public Long getValue() {
        if (firstEventTime == -1) {
            return null;
        } else {
            long timeUnitsPassed = (System.nanoTime() - aggregationIntervalStartTime) / nanosInEventTimeUnit;
            assert timeUnitsPassed >= 0;
            if (timeUnitsPassed > 1) {
                return 0L;
            } else if (timeUnitsPassed == 1) {
                return count;
            } else {
                assert timeUnitsPassed == 0;
                return previousCount;
            }
        }
    }

    @Override
    public String getFormattedValue() {
        Long value = getValue();
        return value == null ? null : value + "/" + TIME_UNIT_NAMES.get(timeUnit);
    }

    @Override
    public String toString() {
        return "ThroughputAggregatorImpl{" +
                getFormattedValue() +
                '}';
    }
}
